use std::sync::Arc;

use async_trait::async_trait;
use serde_json::Value;
use taos::{AsyncFetchable, AsyncQueryable, Itertools, Taos, TryStreamExt};
use tracing::info;

use anyhow::{anyhow, Result};

pub struct TDengineRepo {
    td_conns: Vec<Taos>,
}

impl TDengineRepo {
    pub fn new(dbs: Vec<Taos>) -> Self {
        Self { td_conns: dbs }
    }
    pub fn get_conn(&self, machine_index: u32) -> Result<&Taos> {
        let index = machine_index -1 ;
        self.td_conns.get(index as usize).ok_or(anyhow!("machine_index is {},不符合要求",machine_index))
    }
    pub async fn query(&self, machine_index: u32, query: String) -> Result<Vec<Value>> {
        let conn = self.get_conn(machine_index)?;
        let query_clone = query.clone();
        let result = conn.query(query_clone).await;
        let mut result = match result {
            Ok(result) => result,
            Err(err) => {
                return Err(anyhow!("查询taos 错误 ,sql语句 is {}, err is {}", query, err));
            }
        };

        // print column names
        let meta = result.fields();
        info!("{}", meta.iter().map(|field| field.name()).join("\t"));

        let mut ret = Vec::new();
        // print rows
        let mut rows = result.rows();
        while let Ok(Some(row)) = rows.try_next().await {
            let mut row_map = serde_json::Map::new();
            for (name, value) in row {
                row_map.insert(name.to_string(), value.to_json_value());
            }
            ret.push(Value::Object(row_map));
        }
        Ok(ret)
    }
    pub async fn exec_single(&self ,machine_index:u32,query:String) -> Result<()>{
        let conn = self.get_conn(machine_index)?;
        let query_clone = query.clone();
        let inserted = conn.exec(query_clone).await?;
        Ok(())
    }
}

